package cn.gwm.flink.streaming.function.process;

import cn.gwm.flink.streaming.constant.BaseFields;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.io.IOException;
import java.util.Iterator;
import java.util.Map;

/**
 * 处理gps数据合并到can流中
 * @author GW00256253
 */
public class OdsConnectDataProcessFunction extends ProcessFunction<JSONObject, JSONObject> {

    private OutputTag<JSONObject> gpsTag;
    private OutputTag<JSONObject> gpsHbaseTag;
    private OutputTag<JSONObject> requestTag;

    private transient MapState<String, JSONObject> mapState;
    private transient ListState<JSONObject> listState;
    private transient ListState<JSONObject> tempState;
    private transient MapState<String,Integer> countState;

    private transient ValueState<JSONObject> canObj;
    private transient ValueState<Long> onTimerTs;
//    private transient ValueState<JSONObject> requestObj;

    private String NO_ACTION = "1023";


    public OdsConnectDataProcessFunction(OutputTag<JSONObject> gpsTag, OutputTag<JSONObject> gpsHbaseTag, OutputTag<JSONObject> requestTag){
        this.gpsTag = gpsTag;
        this.gpsHbaseTag = gpsHbaseTag;
        this.requestTag = requestTag;
    }

    @Override
    public void onTimer(long timestamp, ProcessFunction<JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
        super.onTimer(timestamp, ctx, out);
        clearObj(out,ctx);
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.seconds(300))
                .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired).build();
        MapStateDescriptor<String, JSONObject> mapState = new MapStateDescriptor<>("mapState", String.class, JSONObject.class);
        mapState.enableTimeToLive(ttlConfig);
        this.mapState = getRuntimeContext().getMapState(mapState);
        this.canObj = getRuntimeContext().getState(new ValueStateDescriptor<>("canObj",JSONObject.class));
        this.onTimerTs = getRuntimeContext().getState(new ValueStateDescriptor<>("onTimerTs", Long.class));
//        this.requestObj = getRuntimeContext().getState(new ValueStateDescriptor<>("requestObj",JSONObject.class));
        this.countState = getRuntimeContext().getMapState(new MapStateDescriptor<>("countState", String.class, Integer.class));
        this.listState =  getRuntimeContext().getListState(new ListStateDescriptor<JSONObject>("listState",JSONObject.class));
        this.tempState =  getRuntimeContext().getListState(new ListStateDescriptor<JSONObject>("tempState",JSONObject.class));
    }

    @Override
    public void processElement(JSONObject object, ProcessFunction<JSONObject, JSONObject>.Context context, Collector<JSONObject> out) throws Exception {
        if(BaseFields.COMMAND_STATUS.equals(object.getStr(BaseFields.command))){
            setRequestObj(object);
            if(canObj.value()!=null&&onTimerTs.value()!=null){
                context.timerService().deleteProcessingTimeTimer(onTimerTs.value());
                clearObj(out,context);
            }
            if(!mapState.isEmpty()){
                if(mapState.contains(object.getStr(BaseFields.tid))){
                    JSONObject obj = mapState.get(object.getStr(BaseFields.tid));
                    if(obj!=null){
                        extracted(object, obj, 0, out);
                    }else{
                        cacheCanObj(object, context);
                    }
                }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+1000L))){
                    JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+1000L));
                    if(obj!=null){
                        extracted(object, obj, 1, out);
                    }else{
                        cacheCanObj(object, context);
                    }
                }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-1000L))){
                    JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-1000L));
                    if(obj!=null){
                        extracted(object, obj, -1, out);
                    }else{
                        cacheCanObj(object, context);
                    }
                }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+2000L))){
                    JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+2000L));
                    if(obj!=null){
                        extracted(object, obj, 2, out);
                    }else{
                        cacheCanObj(object, context);
                    }
                }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-2000L))){
                    JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-2000L));
                    if(obj!=null){
                        extracted(object, obj, -2, out);
                    }else{
                        cacheCanObj(object, context);
                    }
                }else{
                    cacheCanObj(object, context);
                }
                Iterator<Map.Entry<String, JSONObject>> iterator = mapState.iterator();
                while (iterator.hasNext()){
                    Map.Entry<String, JSONObject> next = iterator.next();
                    if( next!=null && next.getKey().compareTo(object.getStr(BaseFields.tid))<=0){
                        iterator.remove();
                    }
                }
            }else{
                cacheCanObj(object, context);
            }
        }else if(BaseFields.COMMAND_TBOX.equals(object.getStr(BaseFields.command))){
            //gps信号直接输出到侧输出流
            gpsOutPut(object, context);
            if(canObj!=null&&canObj.value()!=null){
                JSONObject canObject = canObj.value();
                if(canObject!=null){
                    if(object.getStr(BaseFields.tid).equals(canObject.getStr(BaseFields.tid))){
                        canObject.set(BaseFields.LONGITUDE, object.getStr(BaseFields.LONGITUDE));
                        canObject.set(BaseFields.LATITUDE, object.getStr(BaseFields.LATITUDE));
                        canObject.set(BaseFields.IS_GPS_FLAG, object.getStr(BaseFields.IS_GPS_FLAG));
                        canObject.set(BaseFields.MATCH_ERROR, 0);
                        out.collect(canObject);
                        canObj.clear();
                        onTimerTs.clear();
                    }else {
                        putGpsObj(object);
                    }
                }else{
                    putGpsObj(object);
                }
            }else{
                putGpsObj(object);
            }
        }else if(BaseFields.COMMAND_REMOTE_REQUEST.equals(object.getStr(BaseFields.command))){
            context.output(requestTag, object);
            boolean bool = false;
            if(object.containsKey(BaseFields.DISCOUNT_POWER)
                    ||object.containsKey(BaseFields.DISCOUNT_TIME)
                    ||object.containsKey(BaseFields.EXPENSIVE_POWER)
                    ||object.containsKey(BaseFields.EXPENSIVE_TIME)) {
                if (NO_ACTION.equals(object.getStr(BaseFields.DISCOUNT_POWER))) {
                    object.set(BaseFields.DISCOUNT_POWER, StrUtil.DASHED);
                } else {
                    bool = true;
                }
                if (NO_ACTION.equals(object.getStr(BaseFields.DISCOUNT_TIME))) {
                    object.set(BaseFields.DISCOUNT_TIME, StrUtil.DASHED);
                } else {
                    bool = true;
                }
                if (NO_ACTION.equals(object.getStr(BaseFields.EXPENSIVE_POWER))) {
                    object.set(BaseFields.EXPENSIVE_POWER, StrUtil.DASHED);
                } else {
                    bool = true;
                }
                if (NO_ACTION.equals(object.getStr(BaseFields.EXPENSIVE_TIME))) {
                    object.set(BaseFields.EXPENSIVE_TIME, StrUtil.DASHED);
                } else {
                    bool = true;
                }
            }
            if(bool){
                listState.add(object);
//                requestObj.update(object);
            }
        }
    }

    private void gpsOutPut(JSONObject object, ProcessFunction<JSONObject, JSONObject>.Context context) {
        context.output(gpsTag, object);
    }

    /**
     * 缓存can周期信号，等待定时器触发
     * @param object
     * @param context
     * @throws IOException
     */
    private void cacheCanObj(JSONObject object, ProcessFunction<JSONObject, JSONObject>.Context context) throws IOException {
        canObj.update(object);
        long ts = context.timerService().currentProcessingTime() + 8000L;
        onTimerTs.update(ts);
        context.timerService().registerProcessingTimeTimer(ts);
    }

    /**
     *  缓存gps数据
     * @param obj
     * @throws Exception
     */
    private void putGpsObj(JSONObject obj) throws Exception {
        JSONObject object = new JSONObject();
        object.set(BaseFields.LONGITUDE, obj.getStr(BaseFields.LONGITUDE));
        object.set(BaseFields.LATITUDE, obj.getStr(BaseFields.LATITUDE));
        object.set(BaseFields.IS_GPS_FLAG, obj.getStr(BaseFields.IS_GPS_FLAG));
        object.set(BaseFields.tid, obj.getStr(BaseFields.tid));
        object.set(BaseFields.vin, obj.getStr(BaseFields.vin));
        mapState.put(object.getStr(BaseFields.tid), object);
    }

    private void setRequestObj(JSONObject object) throws Exception {
        if(listState!=null){
            Iterable<JSONObject> iterable = listState.get();
            if(iterable!=null){
                Iterator<JSONObject> it = iterable.iterator();
                if(it!=null&&it.hasNext()){
                    JSONObject request = it.next();
                    if(request!=null){
                        object.set(BaseFields.DISCOUNT_POWER, request.getStr(BaseFields.DISCOUNT_POWER));
                        object.set(BaseFields.DISCOUNT_TIME, request.getStr(BaseFields.DISCOUNT_TIME));
                        object.set(BaseFields.EXPENSIVE_POWER, request.getStr(BaseFields.EXPENSIVE_POWER));
                        object.set(BaseFields.EXPENSIVE_TIME, request.getStr(BaseFields.EXPENSIVE_TIME));
                        object.set(BaseFields.REQUEST_TIME, request.getStr(BaseFields.REQUEST_TIME));
                        if(countState!=null&&!countState.isEmpty()){
                            Integer integer = countState.get(request.getStr(BaseFields.REQUEST_TIME));
                            if(integer!=null&&integer.intValue()>0){
                                countState.put(request.getStr(BaseFields.REQUEST_TIME),integer.intValue()+1);
                                if(integer.intValue()>10){
                                    countState.remove(request.getStr(BaseFields.REQUEST_TIME));
                                    tempState.clear();
                                    while (it.hasNext()){
                                        tempState.add(it.next());
                                    }
                                    listState.clear();
                                    if(tempState!=null){
                                        Iterable<JSONObject> cacheIt = tempState.get();
                                        if(cacheIt!=null){
                                            Iterator<JSONObject> iterator = cacheIt.iterator();
                                            if(iterator!=null){
                                                while (iterator.hasNext()){
                                                    listState.add(iterator.next());
                                                }
                                                tempState.clear();
                                            }
                                        }
                                    }
                                }
                            }else{
                                countState.put(request.getStr(BaseFields.REQUEST_TIME),1);
                            }
                        }else{
                            countState.put(request.getStr(BaseFields.REQUEST_TIME),1);
                        }
                    }
                }
            }
        }
        /*if(requestObj!=null&&requestObj.value()!=null){
            JSONObject request = requestObj.value();
            if(request!=null){
                object.set(BaseFields.DISCOUNT_POWER, request.getStr(BaseFields.DISCOUNT_POWER));
                object.set(BaseFields.DISCOUNT_TIME, request.getStr(BaseFields.DISCOUNT_TIME));
                object.set(BaseFields.EXPENSIVE_POWER, request.getStr(BaseFields.EXPENSIVE_POWER));
                object.set(BaseFields.EXPENSIVE_TIME, request.getStr(BaseFields.EXPENSIVE_TIME));
                object.set(BaseFields.REQUEST_TIME, request.getStr(BaseFields.REQUEST_TIME));
                if((object.getLong(BaseFields.tid) - request.getLong(BaseFields.tid))>120000){
                    requestObj.clear();
                }
            }
        }*/
    }

    private void clearObj(Collector<JSONObject> out,Context ctx) throws Exception {
        if(canObj!=null&&canObj.value()!=null){
            JSONObject object = canObj.value();
            if(object!=null){
                if(!mapState.isEmpty()){
                    boolean bool = false;
                    if(mapState.contains(object.getStr(BaseFields.tid))){
                        JSONObject obj = mapState.get(object.getStr(BaseFields.tid));
                        if(obj!=null) {
                            extracted(object,obj,0,null);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+1000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+1000L));
                        if(obj!=null) {
                            extracted(object,obj,1,null);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-1000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-1000L));
                        if(obj!=null) {
                            extracted(object,obj,-1,null);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+2000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+2000L));
                        if(obj!=null) {
                            extracted(object,obj,2,null);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-2000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-2000L));
                        if(obj!=null) {
                            extracted(object,obj,-2,null);
                        }
                    }else{
                        bool = true;
                    }
                    Iterator<Map.Entry<String, JSONObject>> iterator = mapState.iterator();
                    while (iterator.hasNext()){
                        Map.Entry<String, JSONObject> next = iterator.next();
                        if(next!=null&&bool){
                            bool = false;
                            JSONObject value = next.getValue();
                            ctx.output(gpsHbaseTag,value);
                        }
                        if( next!=null && (next.getKey().compareTo(object.getStr(BaseFields.tid))<=0)){
                            iterator.remove();
                        }
                    }
                }
                out.collect(object);
                canObj.clear();
                onTimerTs.clear();
            }
        }
    }

    /**
     *  读取gps信号放入can信号中
     * @param object
     * @param obj
     * @param value
     * @param out
     */
    private void extracted(JSONObject object, JSONObject obj, int value, Collector<JSONObject> out) {
        object.set(BaseFields.LONGITUDE, obj.getStr(BaseFields.LONGITUDE));
        object.set(BaseFields.LATITUDE, obj.getStr(BaseFields.LATITUDE));
        object.set(BaseFields.IS_GPS_FLAG, obj.getStr(BaseFields.IS_GPS_FLAG));
        object.set(BaseFields.MATCH_ERROR, value);
        if(null!=out){
            out.collect(object);
        }
    }

}
